package top.cyuw.simplerpc.remoting.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import top.cyuw.simplerpc.config.RpcShutdownHook;
import top.cyuw.simplerpc.constant.RpcConfigKeyConstants;
import top.cyuw.simplerpc.context.RpcContext;
import top.cyuw.simplerpc.extension.ExtensionLoader;
import top.cyuw.simplerpc.registry.ServiceRegistry;
import top.cyuw.simplerpc.registry.ServiceRegistrySelector;
import top.cyuw.simplerpc.remoting.RpcRemoting;
import top.cyuw.simplerpc.remoting.codec.RpcMessageDecoder;
import top.cyuw.simplerpc.remoting.codec.RpcMessageEncoder;
import top.cyuw.simplerpc.util.NamedThreadFactory;

import java.net.InetAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author chen
 * @date 2023/3/13 11:58
 */
@Slf4j
public class SimpleRpcServer implements RpcRemoting {

    private final RpcContext context;
    private final AtomicBoolean started = new AtomicBoolean(false);
    @Getter
    private final ServiceRegistry serviceRegistry;
    private final InvokeHandler invokeHandler;
    private ChannelFuture future;
    private Thread mainThread;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private EventLoopGroup serviceGroup;

    public SimpleRpcServer() {
        context = RpcContext.getInstance();
        String serviceRegistryName = ServiceRegistrySelector.getServiceRegistry();
        serviceRegistry = ExtensionLoader.of(ServiceRegistry.class).getExtension(serviceRegistryName);
        invokeHandler = new InvokeHandler(serviceRegistry);
    }

    @Override
    public void start() {
        if (started.compareAndSet(false, true)) {
            doStart();
        }
    }

    private void doStart() {
        Integer port = context.getRpcConfig().getInteger(RpcConfigKeyConstants.SERVER_PORT, 9988);
        new RpcShutdownHook().register();

        bossGroup = new NioEventLoopGroup(1, new NamedThreadFactory("simplerpc-server-boss"));
        workerGroup = new NioEventLoopGroup(new NamedThreadFactory("simplerpc-server-worker"));
        serviceGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, new NamedThreadFactory("service-executor"));

        mainThread = new Thread(() -> {
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childHandler(new ChannelInitializer<Channel>() {
                            @Override
                            protected void initChannel(Channel ch) {
                                ch.pipeline()
                                        .addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS))
                                        .addLast(new RpcMessageEncoder())
                                        .addLast(new RpcMessageDecoder())
                                        .addLast(serviceGroup, new SimpleRpcServerHandler(invokeHandler));
                            }
                        });

                String host = InetAddress.getLocalHost().getHostAddress();
                future = bootstrap.bind(host, port).sync();
                future.channel().closeFuture().sync();
            } catch (Exception e) {
                log.error("start server failed: " + e.getMessage(), e);
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        });
        mainThread.setName("simplerpc-main");
        mainThread.start();
    }

    @Override
    public void stop() {
        if (started.compareAndSet(true, false)) {
            doStop();
        }
    }

    private void doStop() {
        future.channel().close();

        if (null != bossGroup) {
            bossGroup.shutdownGracefully();
        }

        if (null != workerGroup) {
            workerGroup.shutdownGracefully();
        }

        if (null != serviceGroup) {
            serviceGroup.shutdownGracefully().awaitUninterruptibly();
        }

        if (null != mainThread) {
            mainThread.interrupt();
        }
    }

    @Override
    public RpcContext getContext() {
        return context;
    }
}
